package com.telra.app.server2.cassandra;

import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.OrderProtos;
import com.telra.app.server2.util.DataMapping;
import me.prettyprint.cassandra.model.CqlQuery;
import me.prettyprint.cassandra.model.CqlRows;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.query.QueryResult;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import java.text.DecimalFormat;
import java.util.List;

/**
 * Used to execute on-off queries against Cassandra for request/reply operations.
 */
public class CQLQuery implements CassandraEnabled {

    private final static Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(CQLQuery.class);
    private final static DecimalFormat fmt = new DecimalFormat("#,##0.000000");

    private Keyspace keyspace;

    public CQLQuery() {
        //empty
    }

    public void orderQuery(String orderID) {
        this.executeQuery("select category,itemName,price from Orders where KEY = '" + orderID + "'");
    }

    public void queryByCategory(OrderEvent.Category category, int rowlimit)  {
        this.executeQuery("select category,itemName,price from Orders where category = '" + category + "' limit " + rowlimit, false);
    }

    private void executeQuery(String query, boolean suppressKeyInColumns)  {
        CqlQuery<String, String, String> cql = getCqlQuery(suppressKeyInColumns).setQuery(query);
        log.info("executing query=" + query );
        QueryResult<CqlRows<String, String, String>> cqlResult = cql.execute();
        log.info("done in " + (fmt.format(cqlResult.getExecutionTimeMicro()/1000000.0) ) );
        log.info("size = " + cqlResult.get().getCount() );

        for (Row<String, String, String> row : cqlResult.get()) {

            //String key = row.getKey();
            ColumnSlice<String, String> colSlice = row.getColumnSlice();

            List<HColumn<String, String>> cols = colSlice.getColumns();

            if (cols.size() == 0) {
                return;
            }
            OrderProtos.OrderTicket.Category cat = DataMapping.getCategory(cols.get(0).getValue());

            OrderProtos.OrderTicket order = OrderProtos.OrderTicket.newBuilder()
                    .setItemName(cols.get(1).getValue())
                    .setItemCategory(cat)
                    .setPrice(Double.valueOf(cols.get(2).getValue()))
                    .build();

            log.info(order);

            //for (HColumn col : cols) {
            //    log.info(col.getName() + "=" + col.getValue());
            //}
        }
    }

    private void executeQuery(String query)  {
        executeQuery(query, false);
    }


    private CqlQuery<String, String, String> getCqlQuery(boolean suppressKeyInColumns) {
        return new CqlQuery<String, String, String>(keyspace,
                StringSerializer.get(),   //key
                StringSerializer.get(),   //name
                StringSerializer.get()).setSuppressKeyInColumns(suppressKeyInColumns) ; //value
    }

    public Keyspace getKeyspace() {
        return keyspace;
    }

    public void setKeyspace(Keyspace keyspace) {
        this.keyspace = keyspace;
    }

}
